use std::collections::{HashMap, HashSet};
use std::net::SocketAddr;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
use tokio::sync::{broadcast, Mutex};
use chat_core::message::{Message, MessageType};
use chat_core::utils;
use mem_data::onLineUsers;
use crate::mem_data;
use crate::mem_data::offlineMsgQueue;

pub async fn run(){
    let listener = TcpListener::bind("127.0.0.1:9999").await.unwrap();
    log::info!("Server::run; addr=127.0.0.1:9999");
    log::info!("listening on tcp://127.0.0.1:9999");
    let account_tcp_addr = Arc::clone(&onLineUsers);
    let offline_msg = Arc::clone(&offlineMsgQueue);
    let (tx, _rx) = broadcast::channel(100);
    loop {
        let (mut socket, socket_addr) = listener.accept().await.unwrap();
        log::info!("accepted connection from {}", socket_addr);
        let tx= tx.clone();
        let account_tcp_addr = Arc::clone(&account_tcp_addr);
        let offline_msg = Arc::clone(&offline_msg);
        let mut rx = tx.subscribe();
        tokio::spawn(async move {
            let socket_addr = socket.peer_addr().unwrap();
            let (mut reader, mut writer) = socket.split();
            loop {
                tokio::select! {
                    result = utils::read_message(&mut reader) => {
                        let mut map = account_tcp_addr.lock().await;
                        match result {
                            Ok(message) => {
                                log::info!("server received message from {:?}: {:?}", socket_addr, message);
                                if let None = map.get(&message.sender_account) {
                                    let account = &message.sender_account.clone();
                                    map.insert(account.clone(), socket_addr);
                                    log::info!("new account {} connected", &message.sender_account);
                                    let mut offline_msg = offline_msg.lock().await;
                                    match offline_msg.get_mut(account){
                                        Some(queue) => {
                                            queue.iter().for_each(|msg|{
                                                tx.send((socket_addr, msg.clone())).unwrap();
                                            });
                                            queue.clear();
                                            log::info!("offline message send successfully");
                                        }
                                        None => {
                                            log::info!("offline message queue not found for {:?}", account);
                                        }
                                    }
                                }
                                if let  None = map.get(&message.to_account) {
                                    if message.message_type != MessageType::TextLogin{
                                        let to_account = message.to_account.clone();
                                        let mut offline_msg = offline_msg.lock().await;
                                        match offline_msg.get_mut(&to_account){
                                            Some(queue) => {
                                                queue.insert(message);
                                                log::info!("offline message queue push successfully");
                                            }
                                            None => {
                                                let mut queue = HashSet::new();
                                                queue.insert(message);
                                                offline_msg.insert(to_account, queue);
                                                log::info!("offline message queue push successfully");
                                            }
                                        }
                                    }
                                }else{
                                    tx.send((socket_addr, message)).unwrap();
                                }
                            }
                            Err(e) => {
                                log::error!("read message failed : {:?}",e);
                                socket.shutdown().await.unwrap();
                                map.retain(|_, v| *v != socket_addr);
                                log::info!("shutting down socket {}", socket_addr);
                                log::info!("online sockets {:?}", map);
                                break;
                            }
                        }
                    }
                    result = rx.recv() => {
                        let mut map = account_tcp_addr.lock().await;
                        match result{
                            Ok((addr, message)) => {
                                log::info!("rx received message from {:?}: {:?}", addr, message);
                                let to_account = message.to_account.clone();
                                let to_addr = map.get(&to_account);
                                if  to_addr.is_some() {
                                    if *to_addr.unwrap() == socket_addr {
                                        log::info!("message transfer to {socket_addr:?} successfully");
                                        let _ = utils::send_data(&mut writer, message).await;
                                        log::info!("server send from  message transfer to {socket_addr:?}  successfully")
                                    }

                                }
                            }
                            Err(e) => {
                                log::error!("read message failed : {:?}",e);
                                socket.shutdown().await.unwrap();
                                map.retain(|_, v| *v != socket_addr);
                                log::info!("shutting down socket {}", socket_addr);
                                log::info!("online sockets {:?}", map);
                                break;
                            }
                        }
                    }
                }
            }
        });
    }
}